package com.zhangjianbing.hystrix.custom02.annotation;

import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.stereotype.Service;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author zhangjianbing
 * time 2021/2/20
 */
@Slf4j
@Service
public class HystixCmdImpl implements HystixCmd {

    /**
     * 一个窗口的时间是5
     */
    private static Long WINDOW_SLIDE_TIME = 5000L;

    /**
     * 在一个窗口内失败次数超过20%，熔断
     */
    private static Integer MAX_FAIL_rate = 20;

    /**
     * 锁对象
     */
    private final Object lock = new Object();

    /**
     * 默认是关的
     */
    private HystrixStatus status = HystrixStatus.CLOSE;

    /**
     * 随机数
     */
    private static Random RDM = new Random();

    /**
     * 在一个窗口内失败次数
     */
    private AtomicInteger currentFallCount = new AtomicInteger(0);

    /**
     * 在一个窗口内成功次数
     */
    private AtomicInteger currentSuccessCount = new AtomicInteger(1);

    private ExecutorService executorService = new ThreadPoolExecutor(1, 2, 1L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));

    {
        executorService.execute(() -> {
            while (true) {
                try {
                    Thread.sleep(WINDOW_SLIDE_TIME); // 5s 执行一次检查
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                if (this.getStatus() == HystrixStatus.CLOSE) { // 断路器关闭
                    currentFallCount.set(0);
                    currentSuccessCount.set(1);
                } else {
                    log.info("断路器开启。。。");
                    synchronized (lock) {
                        try {
                            lock.wait();
                            log.info("远程服务正常了，清空窗口数据");
                            currentFallCount.set(0);
                            currentSuccessCount.set(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        });
    }

    /**
     * 拦截请求
     */
    @Override
    public Object interceptor(ProceedingJoinPoint joinPoint) {
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        Method method = methodSignature.getMethod();
        // 现在没法得到接口和实现类
        Object object = getFallCallback(method);
        Method callBackMethod = null;
        try {
            String name = method.getName();
            Class<?>[] parameterTypes = method.getParameterTypes();
            callBackMethod = object.getClass().getMethod(name, parameterTypes);
        } catch (NoSuchMethodException | SecurityException e1) {
            e1.printStackTrace();
        }
        Object fallBackResult = null;
        try {
            // 调用callback方法
            fallBackResult = callBackMethod.invoke(object, joinPoint.getArgs());
        } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
            e.printStackTrace();
        }
        return fallBackResult;
    }

    /**
     * 正常调用
     */
    @Override
    public Object pass(ProceedingJoinPoint request) {
        Object result;
        try {
            result = request.proceed(request.getArgs());
            // 成功了，统计成功次数
            currentSuccessCount.getAndIncrement();
        } catch (Throwable e) {
            log.error("远程接口异常：{}", e.getMessage(), e);
            // 失败了，统计失败次数
            currentFallCount.getAndIncrement();
            if (currentFallCount.get() / currentSuccessCount.get() >= MAX_FAIL_rate) { // 失败超过阈值
                log.info("当前的窗口里面已经达到失败的阈值了，断路器打开");
                this.changeStatus(HystrixStatus.OPEN);// 打开断路器，一段时间后，把断路器改为半开
                executorService.execute(() -> {
                    try {
                        Thread.sleep(5000L);
                        log.info("过了一段时间，把断路器改为半开");
                        this.changeStatus(HystrixStatus.HALF_OPEN);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });
            }
            result = interceptor(request);
        }
        return result;
    }

    @Override
    public Object test(ProceedingJoinPoint request) {
        int num = RDM.nextInt(100) + 1;
        Object result = null;
        if (num <= 3) { // 做测试
            log.info("命中test，去测试接口是否恢复正常");
            try {
                result = request.proceed(request.getArgs());
                // 测试通过,关闭断路器
                this.changeStatus(HystrixStatus.CLOSE);
                // 清空计数
                synchronized (lock) {
                    log.info("远程调用的测试已经通过，远程服务正常");
                    lock.notifyAll();
                }
            } catch (Exception e) {
                log.info("测试没有通过1");
                // e.printStackTrace();
                return interceptor(request);
            } catch (Throwable throwable) {
                log.info("测试没有通过2");
            }
        } else {
            log.info("未进测试，直接调用fallback接口");
            return interceptor(request);
        }
        return result;
    }

    @Override
    public void changeStatus(HystrixStatus status) {
        this.status = status;
    }

    @Override
    public HystrixStatus getStatus() {
        return this.status;
    }

    private Object getFallCallback(Method method) {
        HystixCmdRpc annotation = method.getAnnotation(HystixCmdRpc.class);
        Class<?> callback = annotation.callback();
        try {
            return callback.newInstance();
        } catch (InstantiationException | IllegalAccessException e) {
            e.printStackTrace();
        }
        return null;
    }

}
